#include "SocketServer.h"
#include "IOBuffer.h"
#include "tstring.h"
#include "NamedIndex.h"
#include "ThreadPool.h"
#include "Exception.h"
#include "Utils.h"
using JetByteTools::Win32::CThreadPool;
#include "TangoThreadPool.h"
#include "ICriticalSectionFactory.h"
#include "TangoSocketServer.h"
using JetByteTools::Win32::CIOCompletionPort;
using JetByteTools::Win32::CIOBuffer;
using JetByteTools::Win32::Output;
using JetByteTools::Win32::ToString;
using JetByteTools::Win32::_tstring;
using JetByteTools::Win32::CException;
using JetByteTools::Win32::CCriticalSection;
using JetByteTools::Win32::GetLastErrorMessage;
using JetByteTools::Win32::ICriticalSectionFactory;
using JetByteTools::Win32::CNamedIndex;
typedef JetByteTools::Win32::CSocketServer::IOPool IOPool;
using JetByteTools::Win32::ICriticalSectionFactory;
CTangoSocketServer::CTangoSocketServer(
	const ICriticalSectionFactory &lockFactory,
	IOPool &ioPool,
	unsigned long addressToListenOn,
	unsigned short portToListenOn,
	size_t maxFreeSockets,
	size_t maxFreeBuffers,
	size_t bufferSize,
	size_t numberOfUserDataSlots,
	CTangoThreadPool &businessLogicPool,
	size_t maxMessageSize,
	bool usingReadSequenceNumbers,
	bool usingSequenceNumbers,
	bool issueEarlyRead)
	: JetByteTools::Win32::CSocketServer(lockFactory, ioPool, addressToListenOn, portToListenOn, maxFreeSockets, maxFreeBuffers, bufferSize, numberOfUserDataSlots + 1, usingSequenceNumbers, false),
	m_pool(businessLogicPool),
	m_allocator(maxMessageSize, 10),
	m_maxMessageSize(maxMessageSize),
	m_usingSequenceNumbers(usingReadSequenceNumbers),
	m_issueEarlyRead(issueEarlyRead),
	m_userDataIndex(numberOfUserDataSlots)
{

}

void CTangoSocketServer::OnStartAcceptingConnections()
{
	//global::LogMess(_T("OnStartAcceptingConnections"));
}

void CTangoSocketServer::OnStopAcceptingConnections()
{
	//global::LogMess(_T("OnStopAcceptingConnections"));
}

void CTangoSocketServer::OnShutdownInitiated()
{
	//global::LogMess(_T("OnShutdownInitiated"));
}

void CTangoSocketServer::OnShutdownComplete()
{
	//global::LogMess(_T("OnShutdownComplete"));
}
void CTangoSocketServer::OnConnectionEstablished(
	Socket *pSocket,
	CIOBuffer *pAddress)
{
	//global::LogMess(_T("OnConnectionEstablished"));

	if (0 != GetPerConnectionData(pSocket))
	{
		throw CException(_T("CSocketServer::OnConnectionEstablished()"), _T("Programming error - Socket's server data already contains data"));
	}

	CPerConnectionData *pPerConnectionData = new CPerConnectionData(m_criticalSection, m_allocator);

	SetPerConnectionData(pSocket, pPerConnectionData);

	m_pool.DispatchConnectionEstablished(pSocket, pAddress);
	struct sockaddr_in addrin;
	int len = sizeof(addrin);
	if (!getpeername(pSocket->GetSocket(), (sockaddr*)&addrin, &len))
	{
		pSocket->m_addr = addrin.sin_addr.S_un.S_addr;
		int count = 0;
		//local::inst().AddIpCounts(addrin.sin_addr.S_un.S_addr);
		//local::inst().GetIpCount(addrin.sin_addr.S_un.S_addr, count);
		if (count > 200)
			pSocket->Close();
	}
}

void CTangoSocketServer::OnConnectionClientClose(
	Socket * /*pSocket*/)
{
	//global::LogMess(_T("OnConnectionClientClose"));
}

void CTangoSocketServer::OnConnectionReset(
	Socket * /*pSocket*/)
{
	//global::LogMess(_T("OnConnectionReset"));
}

bool CTangoSocketServer::OnConnectionClosing(
	Socket *pSocket)
{
	//global::LogMess(_T("OnConnectionClosing"));

	m_pool.DispatchConnectionClosing(pSocket);

	return true;      // We'll handle the close on a worker thread 
}

void CTangoSocketServer::OnConnectionClosed(
	Socket * pSocket)
{
	struct sockaddr_in addrin;
	int len = sizeof(addrin);
	//local::inst().DecIpCount(pSocket->m_addr);
	//local::inst().LogOut(pSocket->m_lId, pSocket);
	//global::LogMess("\r\n OnConnectionClosed  ִ����   ( %d )������\r\n", pSocket->m_lId);

	//local::inst().LogOut(NULL, pSocket); //SLogout *pLogout;
	//pSocket->Close();
	//global::LogMess(_T("OnConnectionClosed"));
}

void CTangoSocketServer::OnConnectionCreated()
{
	//global::LogMess(_T("OnConnectionCreated"));
	printf("OnConnectionCreated\r\n");
}

void CTangoSocketServer::OnConnectionDestroyed()
{
	//global::LogMess(_T("OnConnectionDestroyed"));
}

void CTangoSocketServer::OnSocketReleased(
	Socket *pSocket)
{
	//global::LogMess(_T("OnSocketReleased"));
	//global::LogMess("\r\n OnSocketReleasedִ���ˣ����� ( %d ) \r\n", pSocket->m_lId);
	//local::inst().LogOut(pSocket->m_lId,pSocket);
	CPerConnectionData *pPerConnectionData = GetPerConnectionData(pSocket);
	pSocket->m_lId = -1;
	SetPerConnectionData(pSocket, 0);

	delete pPerConnectionData;

	m_pool.OnSocketReleased(pSocket);
}

void CTangoSocketServer::OnConnectionError(
	ConnectionErrorSource source,
	Socket *pSocket,
	CIOBuffer *pBuffer,
	DWORD lastError)
{
	const LPCTSTR errorSource = (source == ZeroByteReadError ? _T(" Zero Byte Read Error:") : (source == ReadError ? _T(" Read Error:") : _T(" Write Error:")));

	//global::LogMess(string(_T("OnConnectionError - Socket = ") + ToString(pSocket) + _T(" Buffer = ") + ToString(pBuffer) + errorSource + GetLastErrorMessage(lastError)).c_str());
}

void CTangoSocketServer::OnError(
	const JetByteTools::Win32::_tstring &message)
{
	//global::LogMess(string(_T("OnError - ") + message).c_str());
}

void CTangoSocketServer::OnBufferCreated()
{
	//global::LogMess(_T("OnBufferCreated"));
}

void CTangoSocketServer::OnBufferAllocated()
{
	// global::LogMess(_T("OnBufferAllocated"));
}

void CTangoSocketServer::OnBufferReleased()
{
	//global::LogMess(_T("OnBufferReleased"));
}

void CTangoSocketServer::OnBufferDestroyed()
{
	//global::LogMess(_T("OnBufferDestroyed"));
}

void CTangoSocketServer::ReadCompleted(
	Socket *pSocket,
	CIOBuffer *pBuffer)
{
	try
	{
		if (m_issueEarlyRead)
		{
			pSocket->Read();
		}

		CPerConnectionData *pPerConnectionData = GetPerConnectionData(pSocket);

		pBuffer->AddRef();

		if (m_usingSequenceNumbers)
		{
			pBuffer = pPerConnectionData->m_outOfSequenceReads.GetNext(pBuffer);
		}

		while (pBuffer)
		{
			//Output(_T("Processing buffer: ") + ToString(pBuffer) + _T(": Seq:") + ToString(pBuffer->GetSequenceNumber()));

			if (!pPerConnectionData->Finished())
			{
				ProcessDataStream(pSocket, pBuffer);
			}

			pBuffer->Release();

			pBuffer = 0;

			if (m_usingSequenceNumbers)
			{
				pBuffer = pPerConnectionData->m_outOfSequenceReads.ProcessAndGetNext();
			}
		}

		if (!m_issueEarlyRead)
		{
			pSocket->Read();
		}
	}
	catch (const CException &e)
	{
		//global::LogMess(string(_T("ReadCompleted - Exception - ") + e.GetWhere() + _T(" - ") + e.GetMessage()).c_str());
		pSocket->Shutdown();
	}
	catch (...)
	{
		//global::LogMess(_T("ReadCompleted - Unexpected exception"));
		pSocket->Shutdown();
	}
}

static size_t AddAtMost(
	CIOBuffer *pBufferOut,
	size_t atMost,
	const CIOBuffer *pBufferIn,
	size_t readOffset)
{
	const size_t available = pBufferIn->GetUsed() - readOffset;

	const size_t bytesToAdd = min(available, atMost);

	pBufferOut->AddData(pBufferIn->GetBuffer() + readOffset, bytesToAdd);

	return bytesToAdd;
}


void CTangoSocketServer::ProcessDataStream(
	Socket *pSocket,
	const CIOBuffer *pBuffer)
{
	CPerConnectionData *pPerConnectionData = GetPerConnectionData(pSocket);

	size_t bufferReadOffset = 0;

	do
	{
		if (!pPerConnectionData->HasBuffer())
		{
			//Output(_T("New buffer"));

			pPerConnectionData->NewBuffer();

			// make sure that we store the message header, even if it's fragmented and we
			// cant actually process it all...

			size_t added = AddAtMost(
				pPerConnectionData->m_pTransferBuffer,
				GetMinimumMessageSize(),
				pBuffer,
				bufferReadOffset);

			bufferReadOffset += added;
		}

		if (pPerConnectionData->m_messageSize == 0)
		{
			// Try and work out the message size.

			if (pPerConnectionData->m_pTransferBuffer->GetUsed() >= GetMinimumMessageSize())
			{
				const size_t messageSize = GetMessageSize(pPerConnectionData->m_pTransferBuffer);

				if (messageSize < GetMinimumMessageSize())
				{
					//global::LogMess(_T("Invalid message size"));

					const std::string response("Invalid message size! Go away!\r\n");

					// Write this message and then shutdown the sending side of the socket.

					pSocket->Write(response.c_str(), response.length());
					pSocket->Shutdown();

					// We cant recover from this. We cant continue to process reads because we'd be out of sync.

					pPerConnectionData->Finished(true);

					break;
				}
				else if (messageSize <= m_maxMessageSize)
				{
					pPerConnectionData->m_messageSize = messageSize;

					//global::LogMess(string(_T("Message size = ") + ToString(messageSize)).c_str());
				}
				else
				{
					//global::LogMess(_T("Too much data!"));

					const std::string response("Too much data! Go away!\r\n");

					// Write this message and then shutdown the sending side of the socket.

					pSocket->Write(response.c_str(), response.length());
					pSocket->Shutdown();

					// We cant recover from this. We cant continue to process reads because we'd be out of sync.

					pPerConnectionData->Finished(true);

					break;
				}
			}
			else
			{
				size_t added = AddAtMost(
					pPerConnectionData->m_pTransferBuffer,
					GetMinimumMessageSize() - pPerConnectionData->m_pTransferBuffer->GetUsed(),
					pBuffer,
					bufferReadOffset);

				bufferReadOffset += added;
			}
		}

		if (pPerConnectionData->m_messageSize != 0)
		{
			// we have a transfer buffer
			// we know how big the message is
			// we know how much of it we already have

			const size_t bytesRequired = pPerConnectionData->m_messageSize - (pPerConnectionData->m_pTransferBuffer->GetUsed() - GetMinimumMessageSize());

			size_t added = AddAtMost(
				pPerConnectionData->m_pTransferBuffer,
				bytesRequired,
				pBuffer,
				bufferReadOffset);

			bufferReadOffset += added;
		}

		if (pPerConnectionData->m_messageSize != 0 && pPerConnectionData->m_messageSize == (pPerConnectionData->m_pTransferBuffer->GetUsed() - GetMinimumMessageSize()))
		{
			// We have a complete message in our transfer buffer.

			//Output(_T("We have a complete message"));

			CIOBuffer *pTransferBuffer = pPerConnectionData->RemoveTransferBuffer();

			if (pTransferBuffer)
			{
				ProcessCommand(pSocket, pTransferBuffer);

				pTransferBuffer->Release();
			}
		}
	} while (bufferReadOffset < pBuffer->GetUsed());
}

size_t CTangoSocketServer::GetMinimumMessageSize() const
{
	// The smallest possible packet is 2 bytes for the message length with a 0 length message.
	return 2;
}

size_t CTangoSocketServer::GetMessageSize(
	const CIOBuffer *pBuffer) const
{
	const BYTE *pData = pBuffer->GetBuffer();

	return *pData + (*(pData + 1) * 256);
}

void CTangoSocketServer::ProcessCommand(
	Socket *pSocket,
	CIOBuffer *pBuffer)
{
	m_pool.DispatchReadCompleted(pSocket, pBuffer);
}

void CTangoSocketServer::SetPerConnectionData(
	Socket *pSocket,
	CPerConnectionData *pData)
{
	pSocket->SetUserPtr(m_userDataIndex, pData);
}

CTangoSocketServer::CPerConnectionData *CTangoSocketServer::GetPerConnectionData(
	const Socket *pSocket)
{
	return (CPerConnectionData*)pSocket->GetUserPtr(m_userDataIndex);
}

CTangoSocketServer::CPerConnectionData::CPerConnectionData(
	CCriticalSection &criticalSection,
	CIOBuffer::Allocator &allocator)
	: m_messageSize(0),
	m_pTransferBuffer(0),
	m_outOfSequenceReads(criticalSection),
	m_transferSequenceNumber(0),
	m_finished(0),
	m_allocator(allocator)
{
}

CTangoSocketServer::CPerConnectionData::~CPerConnectionData()
{
	try
	{
		if (m_pTransferBuffer)
		{
			m_pTransferBuffer->Release();
			m_pTransferBuffer = 0;
		}
	}
	catch (...)
	{
	}
}

void CTangoSocketServer::CPerConnectionData::NewBuffer()
{
	if (m_pTransferBuffer)
	{
		throw "TODO";
	}

	m_messageSize = 0;
	m_pTransferBuffer = m_allocator.Allocate();

	m_pTransferBuffer->SetSequenceNumber(::InterlockedExchangeAdd(&m_transferSequenceNumber, 1));
}

void CTangoSocketServer::CPerConnectionData::AddData(
	const BYTE *pData,
	size_t dataLength)
{
	if (!m_pTransferBuffer)
	{
		throw "TODO";
	}

	m_pTransferBuffer->AddData(pData, dataLength);
}

bool CTangoSocketServer::CPerConnectionData::HasBuffer() const
{
	return (0 != m_pTransferBuffer);
}

CIOBuffer *CTangoSocketServer::CPerConnectionData::RemoveTransferBuffer()
{
	CIOBuffer *pBuffer = m_pTransferBuffer;

	m_pTransferBuffer = 0;
	m_messageSize = 0;

	return pBuffer;
}


bool CTangoSocketServer::CPerConnectionData::Finished()
{
	return (1 == ::InterlockedExchange(&m_finished, m_finished));
}

void CTangoSocketServer::CPerConnectionData::Finished(bool finished)
{
	//lint -e{534} Ignoring return value of function 
	::InterlockedExchange(&m_finished, finished ? 1 : 0);
}

void CTangoSocketServer::RePostCmd(Socket *pSocket,
	JetByteTools::Win32::CIOBuffer *pBuffer)
{

	ProcessCommand(pSocket, pBuffer);
}
